/*
 * RMCommitmentAction.java   0.0.1    9 Nov 2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transactions.management.rm;

import java.lang.reflect.Proxy;
import java.util.List;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.cache.OIDAlreadyInUseException;
import org.mentalsmash.tazio.cache.ServerCache;
import org.mentalsmash.tazio.cache.impl.CachingProxy;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.transactions.RollbackException;
import org.mentalsmash.tazio.transactions.TransactionStatus;
import org.mentalsmash.tazio.transactions.TransactionalSystemRuntimeException;
import org.mentalsmash.tazio.transactions.management.CommitmentAction;
import org.mentalsmash.tazio.transactions.management.RootSetUpdate;
import org.mentalsmash.tazio.transactions.management.TazioTransaction;
import org.mentalsmash.tazio.transactions.management.TransactionStateException;
import org.mentalsmash.tazio.transactions.management.TransactionalOperation;
import org.mentalsmash.tazio.transactions.management.ValidationAgent;
import org.mentalsmash.tazio.transactions.objects.RootSet;
import org.mentalsmash.tazio.transactions.objects.TransactionalObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Andrea Reale
 * @version 0.0.1
 * @synch No non-final status variables + an instance of this class should never
 *        be accessed by more then one thread. No concurrency concerns.
 */
@Concurrency(concurrentAccess = false, specification = "No non-final status variables + an instance of this class should never "
	+ "be accessed by more then one thread. No concurrency concerns.")
final class RMCommitmentAction extends CommitmentAction {

    final static Logger log = LoggerFactory.getLogger(RMCommitmentAction.class);

    private final RMTransactionManager tm;

    private final ValidationAgent validationAgent;

    private final ServerCache cache;

    public RMCommitmentAction(final TazioTransaction toCommit,
	    final RMTransactionManager tm, final RootSet rs,
	    final ServerCache cache) {
	super(toCommit, rs);
	this.tm = tm;
	this.validationAgent = tm.getValidationAgent();
	this.cache = cache;
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.util.concurrent.Callable#call()
     */
    @Override
    public TransactionStatus call() throws Exception {

	doValidateAndCommmit();
	return this.toCommit.getStatus();

    }

    /**
     * @throws TransactionStateException
     * @throws RollbackException
     */
    private void doValidateAndCommmit() throws TransactionStateException,
	    RollbackException {
	if (this.toCommit.getStatus() != TransactionStatus.GLOBAL_VALIDATION) {
	    log.error("Transaction" + this.toCommit + " that wasn't in the "
		    + "GLOBAL_VALIDATIONS status was asked to commit");
	    throw new TransactionStateException("Transaction " + this.toCommit
		    + " was not in the GLOBAL_VALIDATION status");
	}

	if (this.toCommit.isRollbackOnly()) {
	    this.toCommit.stateTransition(TransactionStatus.ROLLED_BACK);
	    log.debug("Transaction {} rolled back", this.toCommit);
	    throw new RollbackException("Transaction " + this.toCommit
		    + " was set to rollback only");
	}

	log.debug("{} started its GLOBAL_VALIDATION", this.toCommit);

	if (!this.validationAgent.validate(this.toCommit)) {
	    log.debug("VALIDATION FAILED");

	    this.toCommit.stateTransition(TransactionStatus.ROLLED_BACK);
	    log.debug("Transaction {} rolled back", this.toCommit);
	    throw new RollbackException(
		    "Global validation failed. Transaction " + this.toCommit
			    + " rolled back");
	}

	log.debug("{} passed the global validation", this.toCommit);

	log.debug("{} is applying its changes to the root set", this.toCommit);

	tm.lockCounter();
	try {
	    this.toCommit.setSeqNo(this.tm.incrementAndGetSeqNoCtr());

	    doApplyChanges();
	    updateCache();

	    log.debug("{}. After committing, before state transition",
		    this.toCommit);
	    
//	    try {
//		log.debug("Waiting before notifying commit of transaction "+toCommit.getTransactionId());
//		Thread.sleep(1000);
//		log.debug("Notifying commit of transaction "+toCommit.getTransactionId());
//	    } catch (InterruptedException e) {
//		// TODO Auto-generated catch block
//		e.printStackTrace();
//	    }
	    
	    this.toCommit.stateTransition(TransactionStatus.COMMITTED);
	} finally {
	    tm.unlockCounter();
	}

	log.debug("{} commited", this.toCommit);

    }

    private void updateCache() {
	for (final OID written : this.toCommit.getWriteSet()) {
	    TransactionalObject obj = super.rs.getObject(written);

	    if (written.equals(RootSet.ROOT_SET_OID)) {
		final List<TransactionalOperation> opsOnRS = this.toCommit
			.getOperations(written);
		
		for (TransactionalOperation op : opsOnRS) {
		    final RootSetUpdate rsu = (RootSetUpdate) op;
		    if (rsu.isRemoval()) {
			this.cache.removeObject((OID) rsu.getValue());
		    } else {
			TransactionalObject toAdd = (TransactionalObject) rsu
				.getValue();
			
			CachingProxy target = (CachingProxy) Proxy.getInvocationHandler(toAdd);
			
			try {
			    this.cache.addObject(toAdd.getObjectIdentifier(),
				    target.getCurrent());
			} catch (OIDAlreadyInUseException e) {
			    log.error("OIDs are broken!!");
			    throw new TransactionalSystemRuntimeException(e);
			}
		    }
		}

		this.cache.update(RootSet.ROOT_SET_OID, super.rs);
	    } else {
		CachingProxy target = (CachingProxy) Proxy.getInvocationHandler(obj);
		this.cache.update(written, target.getCurrent());
	    }
	}
    }

}
